package cn.itcast.streaming.function.flatmap;

import cn.itcast.streaming.bean.OnlineDataObj;
import cn.itcast.streaming.bean.VehicleInfoModel;
import org.apache.flink.streaming.api.functions.co.RichCoFlatMapFunction;
import org.apache.flink.util.Collector;

import java.util.HashMap;

//在线实时故障分析业务开发拉宽车辆基础表的数据
public class VehicleInfoMysqlFunction extends RichCoFlatMapFunction<OnlineDataObj, HashMap<String, VehicleInfoModel>,OnlineDataObj> {
    private HashMap<String,VehicleInfoModel> vehicleInfoModelHashMap = new HashMap<>();
    @Override
    public void flatMap1(OnlineDataObj onlineDataObj, Collector<OnlineDataObj> collector) throws Exception {
        //根据原始数据的vin获取到广播流中的车辆基础信息数据
        VehicleInfoModel vehicleInfoModel = vehicleInfoModelHashMap.getOrDefault(onlineDataObj.getVin(), null);
        if(vehicleInfoModel!=null){
            onlineDataObj.setSeriesName(vehicleInfoModel.getSeriesName()); //车系名称
            onlineDataObj.setModelName(vehicleInfoModel.getModelName());  //车型名称
            onlineDataObj.setLiveTime(vehicleInfoModel.getLiveTime());   //车辆年限
            onlineDataObj.setSalesDate(vehicleInfoModel.getSalesDate()); //销售日期
            onlineDataObj.setCarType(vehicleInfoModel.getCarType());   //车辆类型
            collector.collect(onlineDataObj);
        }else {
            System.out.println("原始数据的vin"+onlineDataObj.getVin()+"在车辆基础信息表中不存在。");
        }
    }


    @Override
    public void flatMap2(HashMap<String, VehicleInfoModel> stringVehicleInfoModelHashMap, Collector<OnlineDataObj> collector) throws Exception {
        vehicleInfoModelHashMap = stringVehicleInfoModelHashMap;
    }
}
